package bloop.bsp

import scala.concurrent.duration.FiniteDuration

import bloop.cli.BspProtocol
import bloop.cli.ExitStatus
import bloop.io.Environment.lineSeparator
import bloop.logging.BspClientLogger
import bloop.logging.RecordingLogger
import bloop.task.Task
import bloop.util.TestProject
import bloop.util.TestUtil

import monix.execution.ExecutionModel
import monix.execution.Scheduler

object TcpBspConnectionSpec extends BspConnectionSpec(BspProtocol.Tcp)
object LocalBspConnectionSpec extends BspConnectionSpec(BspProtocol.Local)

class BspConnectionSpec(
    override val protocol: BspProtocol
) extends BspBaseSuite {
  // A custom pool we use to make sure we don't block on threads
  val poolFor6Clients: Scheduler = Scheduler(
    java.util.concurrent.Executors.newFixedThreadPool(20),
    ExecutionModel.Default
  )

  test("reconnect-noop") {
    TestUtil.withinWorkspace { workspace =>
      val sources = List(
        """/main/scala/Foo.scala
          |class Foo
          """.stripMargin
      )
      val `A` = TestProject(workspace, "a", sources)
      val `B` = TestProject(workspace, "b", sources, List(`A`))
      val projects = List(`A`, `B`)

      val configDir = TestProject.populateWorkspace(workspace, projects)
      val logger = new RecordingLogger(ansiCodesSupported = false)
      val bspLogger = new BspClientLogger(logger)

      def bspCommand() = createBspCommand(configDir)
      val state = TestUtil.loadTestProject(configDir.underlying, logger)
      val bspState = openBspConnection(
        state,
        bspCommand,
        configDir,
        bspLogger
      )
      var lastCompilationHashes: Option[Map[String, Int]] = None
      bspState.withinSession { state =>
        val compiledState = state.compile(`B`)
        assertExitStatus(compiledState, ExitStatus.Ok)
        assert(compiledState.wasLastCompilationNoop.exists(_ == false))
        lastCompilationHashes = compiledState.lastCompilationHashes
        val secondCompiledState = compiledState.compile(`B`)
        assertExitStatus(secondCompiledState, ExitStatus.Ok)
        assert(secondCompiledState.wasLastCompilationNoop.exists(_ == true))
        assertEquals(secondCompiledState.lastCompilationHashes, lastCompilationHashes)
        ()
      }

      val bspState2 = openBspConnection(
        state,
        bspCommand,
        configDir,
        bspLogger
      )

      bspState2.withinSession { state =>
        val compiledState = state.compile(`B`)
        assertExitStatus(compiledState, ExitStatus.Ok)
        assert(compiledState.wasLastCompilationNoop.exists(_ == true))
        assertEquals(compiledState.lastCompilationHashes, lastCompilationHashes)
        val secondCompiledState = compiledState.compile(`B`)
        assertExitStatus(secondCompiledState, ExitStatus.Ok)
        assert(secondCompiledState.wasLastCompilationNoop.exists(_ == true))
        assertEquals(compiledState.lastCompilationHashes, lastCompilationHashes)
        ()
      }

    }
  }

  test("initialize several clients concurrently and simulate a hard disconnection") {
    TestUtil.withinWorkspace { workspace =>
      val `A` = TestProject(workspace, "a", Nil)
      val projects = List(`A`)
      val configDir = TestProject.populateWorkspace(workspace, projects)

      def createClient: Task[Unit] = {
        Task {
          val logger = new RecordingLogger(ansiCodesSupported = false)
          val bspLogger = new BspClientLogger(logger)
          def bspCommand() = createBspCommand(configDir)
          val state = TestUtil.loadTestProject(configDir.underlying, logger)

          // Run the clients on our own unbounded IO scheduler to allow client concurrency
          val scheduler = Some(poolFor6Clients)
          val bspState = openBspConnection(
            state,
            bspCommand,
            configDir,
            bspLogger,
            userIOScheduler = scheduler
          )

          assert(bspState.status == ExitStatus.Ok)
          // Note we opened an unmanaged state and we exit without sending `shutdown`/`exit`
          checkConnectionIsInitialized(logger)

          // Wait 500ms to let the rest of clients to connect to server and then simulate dropout
          Thread.sleep(500)
          bspState.simulateClientDroppingOut()
        }
      }

      val client1 = createClient
      val client2 = createClient
      val client3 = createClient
      val client4 = createClient
      val client5 = createClient
      val client6 = createClient

      // A pool of 20 threads can only support 6 clients concurrently (more will make it fail)
      val firstBatchOfClients = List(client1, client2, client3, client4, client5, client6)
      TestUtil.await(FiniteDuration(5, "s"), poolFor6Clients) {
        Task.gatherUnordered(firstBatchOfClients).map(_ => ())
      }

      val client7 = createClient
      val client8 = createClient
      val client9 = createClient
      val client10 = createClient
      val client11 = createClient
      val client12 = createClient

      // If we can connect six more clients, it means resources were correctly cleaned up
      val secondBatchOfClients = List(client7, client8, client9, client10, client11, client12)
      TestUtil.await(FiniteDuration(5, "s"), poolFor6Clients) {
        Task.gatherUnordered(secondBatchOfClients).map(_ => ())
      }
    }
  }

  def checkConnectionIsInitialized(logger: RecordingLogger): Unit = {
    val contentLogs = logger.debugs.flatMap(_.split("\n")).filter(_.startsWith("  --> content:"))
    // Filter out the initialize request that contains platform-specific details
    val allButInitializeRequest = contentLogs.filterNot(msg =>
      msg.contains("""build/initialize"""") || msg.contains("logMessage")
    )
    assertNoDiff(
      allButInitializeRequest.mkString(lineSeparator),
      s"""|
          |  --> content: ${TestConstants.buildInitialize}
          |  --> content: {"method":"build/initialized","jsonrpc":"2.0"}""".stripMargin
    )
  }

  val poolFor1Client: Scheduler = Scheduler(
    java.util.concurrent.Executors.newFixedThreadPool(6),
    ExecutionModel.Default
  )

  test("trigger slow compilation and simulate hard BSP client disconnection") {
    TestUtil.withinWorkspace { workspace =>
      object Sources {
        val `A.scala` =
          """/A.scala
            |package macros
            |
            |import scala.reflect.macros.blackbox.Context
            |import scala.language.experimental.macros
            |
            |object SleepMacro {
            |  def sleep(): Unit = macro sleepImpl
            |  def sleepImpl(c: Context)(): c.Expr[Unit] = {
            |    import c.universe._
            |    Thread.sleep(1000)
            |    reify { () }
            |  }
            |}""".stripMargin

        val `B.scala` =
          """/B.scala
            |object B { def foo(s: String): String = s.toString; macros.SleepMacro.sleep() }
          """.stripMargin

        val `B2.scala` =
          """/B2.scala
            |object B2 { def foo(s: String): String = s.toString; macros.SleepMacro.sleep() }
          """.stripMargin

        val `B3.scala` =
          """/B3.scala
            |object B3 { def foo(s: String): String = s.toString; macros.SleepMacro.sleep() }
          """.stripMargin

        val `B4.scala` =
          """/B4.scala
            |object B4 { def foo(s: String): String = s.toString; macros.SleepMacro.sleep() }
          """.stripMargin

        val `B5.scala` =
          """/B5.scala
            |object B5 { def foo(s: String): String = s.toString; macros.SleepMacro.sleep() }
          """.stripMargin

        val `B6.scala` =
          """/B6.scala
            |object B6 { def foo(s: String): String = s.toString; macros.SleepMacro.sleep() }
          """.stripMargin
      }

      val `A` = TestProject(workspace, "a", List(Sources.`A.scala`))
      val sourcesB = List(
        Sources.`B.scala`,
        Sources.`B2.scala`,
        Sources.`B3.scala`,
        Sources.`B4.scala`,
        Sources.`B5.scala`,
        Sources.`B6.scala`
      )

      val `B` = TestProject(workspace, "b", sourcesB, List(`A`))

      val projects = List(`A`, `B`)
      val configDir = TestProject.populateWorkspace(workspace, projects)
      val logger = new RecordingLogger(ansiCodesSupported = false)

      def createHangingCompilationViaBsp: Task[Unit] = {
        Task {
          val bspLogger = new BspClientLogger(logger)
          def bspCommand() = createBspCommand(configDir)
          val state = TestUtil.loadTestProject(configDir.underlying, logger)

          // Run the clients on our own unbounded IO scheduler to allow client concurrency
          val scheduler = Some(poolFor1Client)
          val unmanagedBspState = openBspConnection(
            state,
            bspCommand,
            configDir,
            bspLogger,
            userIOScheduler = scheduler
          )

          assert(unmanagedBspState.status == ExitStatus.Ok)
          val bspState = unmanagedBspState.toUnsafeManagedState

          // Note we opened an unmanaged state and we exit without sending `shutdown`/`exit`
          checkConnectionIsInitialized(logger)

          import bloop.engine.ExecutionContext
          import java.util.concurrent.TimeUnit
          ExecutionContext.ioScheduler.scheduleOnce(
            2000,
            TimeUnit.MILLISECONDS,
            new Runnable {
              def run(): Unit = {
                // Simulate hard disconnection by closing streams, should trigger cancellation
                unmanagedBspState.simulateClientDroppingOut()
              }
            }
          )

          // A hand-made request to trigger a compile and detach from its response
          bspState.runAfterTargets(`B`) { target =>
            import ch.epfl.scala.bsp
            import ch.epfl.scala.bsp.endpoints.BuildTarget

            val req = bspState.client0.request(
              BuildTarget.compile,
              bsp.CompileParams(List(target), None, None)
            )
            req.runAsync(ExecutionContext.ioScheduler)
            // Wait until observable is completed, which means server is done
            Task.liftMonixTaskUncancellable(
              bspState.serverStates.foreachL(_ => ())
            )
          }
        }.flatten
      }

      // Time out is 5 to check cancellation works and we don't finish until end
      // of compilation, which would take more than 6 seconds as there are 12 sleeps
      val safeDelay = FiniteDuration(5, "s")
      TestUtil.await(safeDelay, poolFor1Client)(createHangingCompilationViaBsp)
    }
  }
}
